package com.nulldev.util.concurrency.threadIt.v4.executors.dcte;

import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import com.nulldev.util.JVM.Arguments;
import com.nulldev.util.JVM.JVM;
import com.nulldev.util.VariableAPI.ArrayUtils;
import com.nulldev.util.VariableAPI.MathUtil;
import com.nulldev.util.concurrency.threadIt.v4.emapi.ExecutionPreference;
import com.nulldev.util.concurrency.threadIt.v4.emapi.compat.ExecutableFuture;
import com.nulldev.util.concurrency.threadIt.v4.emapi.compat.ExecutableRunnableFuture;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExecutorFeatureExtensions;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExtendedExecutor;
import com.nulldev.util.concurrency.threadIt.v4.utils.impls.EnhancedArrayBlockingQueue;
import com.nulldev.util.concurrency.threadIt.v4.utils.impls.TEConditions;
import com.nulldev.util.data.Arrays.arrays.FastArrayList;
import com.nulldev.util.data.Arrays.arrays.FastLists;
import com.nulldev.util.data.Arrays.misc.CircularIterator;
import com.nulldev.util.io.FastIO.FastIO;

/**
 * Dynamic Chain(-Executed) Thread Executor <br>
 * /// STAGING /// <br>
 * <br>
 * Designed to be simple and capable.
 * 
 * @author null
 * @since 3.0.1
 */
public class DCTE extends AbstractExecutorService implements Runnable, ExtendedExecutor {

	public static final int MAX_CHAIN_LENGTH = 8;
	public static final boolean ENABLE_CHAIN_LIMITS = Arguments.hasArgument("--DCTE.chainLimits");

	public static class Submission implements Runnable {
		public final Runnable r;
		public final AtomicReference<Submission> chain = new AtomicReference<Submission>();

		public Submission(final Runnable r) {
			this.r = r;
		}

		@Override
		public void run() {
			if (this.r != null)
				this.r.run();
		}
	}

	public static class DCTEThread extends Thread {

		private final DCTE inst;
		protected final BlockingQueue<Submission> localQueue = new EnhancedArrayBlockingQueue<Submission>(1);
		private final AtomicBoolean BUSY = new AtomicBoolean();

		public DCTEThread(final DCTE instance, final int idx) {
			super("DCTE[idx=" + idx + "]");
			this.inst = instance;
		}

		@Override
		public void run() {
			while (!this.inst.STOP.get()) {
				Submission s;
				try {
					s = this.localQueue.take();
				} catch (InterruptedException e) {
					continue;
				}
				if (s == null) {
					Thread.yield();
					continue;
				}
				this.BUSY.lazySet(true);
				Submission c = s;
				while (c != null) {
//					System.out.println("Running: " + c + ", Next: " + c.chain.get());
					try {
						c.run();
					} catch (Throwable t) {
						JVM.errorln(t);
					}
					if (c.chain.get() == null) {
						break;
					} else
						c = c.chain.get();
				}
				this.BUSY.lazySet(false);
				Thread.yield();
			}
		}

		public boolean addToQueue(final Submission r) {
			if (r == null)
				return false;
			if (this.BUSY.get() || !this.localQueue.isEmpty())
				return false;
			try {
				return this.localQueue.offer(r);
			} catch (Exception e) {
				FastIO.get().printlnError(e);
				return false;
			}
		}

		public boolean isBusy() {
			return this.BUSY.get();
		}
	}

	public DCTE() {
		this(4);
	}

	public DCTE(final int threadCount) {
		if (threadCount <= 0)
			throw new IllegalArgumentException("threadCount <= 0!");
		for (int i = 0; i < threadCount; i++) {
			final DCTEThread t = new DCTEThread(this, i);
			threads.add(t);
			t.start();
		}
		new Thread(this, "DCTE[Manager]").start();
	}

	@Override
	public String toString() {
		return "DCTE[hash=" + MathUtil.toHex(super.hashCode()) + ",threads=" + this.threads.size() + "]";
	}

	private final BlockingQueue<Submission> queue = new LinkedTransferQueue<Submission>();
	private final List<DCTEThread> threads = new FastArrayList<DCTEThread>(true);

	private final AtomicBoolean STOP = new AtomicBoolean();

	@Override
	public void shutdown() {
		this.STOP.lazySet(true);
		this.queue.clear();
	}

	@Override
	public List<Runnable> shutdownNow() {
		this.STOP.lazySet(true);
		final List<Runnable> out = FastLists.list(this.queue.size());
		out.addAll(this.queue);
		this.queue.clear();
		return out;
	}

	@Override
	public boolean isShutdown() {
		return this.STOP.get();
	}

	@Override
	public boolean isTerminated() {
		return this.STOP.get() && this.queue.isEmpty();
	}

	@Override
	public boolean awaitTermination(final long timeout, final TimeUnit unit) throws InterruptedException {
		this.shutdown();
		return true;
	}

	@Override
	public void execute(final Runnable command) {
		if (command == null)
			return;
		try {
			this.queue.put(new Submission(command));
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public void run() {
		final Iterator<DCTEThread> threads = new CircularIterator<DCTEThread>(this.threads);
		while (!this.STOP.get()) {
			try {
				final Submission sub = this.queue.take();
				if (sub == null)
					continue;
				DCTEThread thread = null;
				for (int i = 0; i < this.threads.size(); i++) {
					final DCTEThread candidate = threads.next();
					if (candidate == null || candidate.isBusy())
						continue;
					thread = candidate;
					break;
				}
				if (thread == null) {
					this.queue.put(sub);
					continue;
				}
				if (!thread.addToQueue(sub)) {
					this.queue.put(sub);
					continue;
				}
			} catch (Throwable tx) {
				tx.printStackTrace();
			}
		}
	}

	private static final ExecutorFeatureExtensions[] SUPPORTED = new ExecutorFeatureExtensions[]
		{ ExecutorFeatureExtensions.EEX_executor_native_batching, ExecutorFeatureExtensions.EEX_query_batching_supported,
				ExecutorFeatureExtensions.EEX_query_slots_supported, ExecutorFeatureExtensions.EEX_core_drain_task_queue,
				ExecutorFeatureExtensions.EEX_query_thread_affinity_supported, ExecutorFeatureExtensions.EEX_compat_native_concurrency_api,
				ExecutorFeatureExtensions.EEX_query_executor_forking, ExecutorFeatureExtensions.EEX_core_execute_with_preference,
				ExecutorFeatureExtensions.EEX_query_debug };

	@Override
	public boolean supportsFeature(final ExecutorFeatureExtensions feature) {
		return ArrayUtils.contains(SUPPORTED, feature);
	}

	@Override
	public ExecutorFeatureExtensions[] supported() {
		return SUPPORTED;
	}

	@Override
	public void batch(final Runnable f) {
		if (f == null)
			return;
		this.execute(f);
	}

	@Override
	public void batchAll(final Runnable[] _r) {
		if (_r == null || _r.length == 0)
			return;
		final boolean requireChainLimits = (_r.length > MAX_CHAIN_LENGTH) && ENABLE_CHAIN_LIMITS;
		Submission current = null, original = null;
		int chain_length = 1;
		for (final Runnable r : _r) {
			if (r == null)
				continue;
			if (current == null && original == null) {
				original = current = new Submission(r);
			} else {
				final Submission e = new Submission(r);
				current.chain.set(e);
				current = e;
				chain_length++;
				if (chain_length > MAX_CHAIN_LENGTH && requireChainLimits) {
					chain_length = 1;
					if (original != null)
						this.queue.offer(original);
					current = null;
					original = null;
				}
			}
		}
		if (original != null && !requireChainLimits)
			this.queue.offer(original);
	}

	/* EEX_core_drain_task_queue */

	@Override
	public void executor_drainTasks(final Queue<Runnable> async_queue, final int batchSize) {
		if (async_queue == null || async_queue.isEmpty())
			return;
		if (batchSize <= 0)
			return;
		final boolean requireChainLimits = (batchSize > MAX_CHAIN_LENGTH) && ENABLE_CHAIN_LIMITS;
		Submission original = null;
		final AtomicReference<Submission> current = new AtomicReference<Submission>();
		int chain_length = 1;
		for (int i = 0; i < batchSize; i++) {
			final Runnable r;
			try {
				r = async_queue.remove();
			} catch (Throwable t) {
				continue;
			}
			if (r == null)
				continue;
			if (current.get() == null && original == null) {
				original = current.updateAndGet((e) -> {
					return new Submission(r);
				});
			} else {
				final Submission e = new Submission(r);
				current.get().chain.set(e);
				current.set(e);
				chain_length++;
				if (chain_length > MAX_CHAIN_LENGTH && requireChainLimits) {
					chain_length = 1;
					if (original != null)
						this.queue.offer(original);
					current.set(null);
					original = null;
				}
			}
		}
		if (original != null && !requireChainLimits)
			this.queue.offer(original);
	}

	/* EEX_query_thread_affinity_supported */

	@Override
	public boolean executorSupportsThreadAffinity() {
		return false;
	}

	@Override
	public boolean executorSupportsThreadPinning() {
		return false;
	}

	/* EEX_query_executor_forking */

	@Override
	public boolean executorIsForking() {
		return false;
	}

	/* EEX_compat_native_concurrency_api */

	@Override
	public Future<?> submit(final Runnable task) {
		TEConditions.validRunnable(task);
		final ExecutableRunnableFuture<Object> ef = new ExecutableRunnableFuture<Object>(task, null);
		execute(ef);
		return ef;
	}

	@Override
	public <T> Future<T> submit(final Runnable task, final T result) {
		TEConditions.validRunnable(task);
		final ExecutableRunnableFuture<T> ef = new ExecutableRunnableFuture<T>(task, result);
		execute(ef);
		return ef;
	}

	@Override
	public <T> Future<T> submit(final Callable<T> task) {
		final ExecutableFuture<T> ef = new ExecutableFuture<T>(task);
		execute(ef);
		return ef;
	}

	@Override
	protected <T> RunnableFuture<T> newTaskFor(final Callable<T> callable) {
		return new ExecutableFuture<T>(callable);
	}

	@Override
	protected <T> RunnableFuture<T> newTaskFor(final Runnable runnable, final T value) {
		TEConditions.validRunnable(runnable);
		return new ExecutableRunnableFuture<T>(runnable, value);
	}

	@Override
	public void execute(final Runnable command, final ExecutionPreference e) {
		if (command == null)
			return;
		this.execute(command);
	}

	/* EEX_query_batching_supported */

	@Override
	public boolean executorSupportsBatching() {
		return false;
	}

	/* EEX_query_slots_supported */

	@Override
	public boolean executorSupportsSlotCount() {
		return false;
	}

	/* EEX_query_debug */

	@Override
	public Object debug_query(final EEXDebugQueryType type) {
		switch (type) {
			case EEX_DEBUG_TOTAL_THREAD_COUNT:
				return this.threads.size();
			case EEX_DEBUG_TASKS_IN_QUEUE:
				return this.queue.size();
			/* */
			case EEX_DCTE_NON_STANDARD_BATCHING_CHAIN_SIZE: {
				if (ENABLE_CHAIN_LIMITS)
					return MAX_CHAIN_LENGTH;
				else
					return Integer.MAX_VALUE;
			}
			/* */
			default:
				return null;
		}
	}
}
